<?php

namespace Lambq\Mqtt\MqttClass;

use Simps\MQTT\Client;
use Simps\MQTT\Protocol\Types;
use Simps\MQTT\Config\ClientConfig;

class Mqtt
{
    protected $host = null;
    protected $username = null;
    protected $cert_file = null;
    protected $local_cert = null;
    protected $local_pk = null;
    protected $password = null;
    protected $port = null;
    protected $timeout = 0;
    protected $debug = null;
    protected $qos = 0;
    protected $retain = 0;
    protected $clientId;
    protected $clean;
    protected $PublishID;
    protected $SubscribeID;

    public function __construct()
    {
        $this->host         = config('mqtt.host');
        $this->username     = config('mqtt.username');
        $this->password     = config('mqtt.password');
        $this->cert_file    = config('mqtt.certfile');
        $this->local_cert   = config('mqtt.localcert');
        $this->local_pk     = config('mqtt.localpk');
        $this->port         = config('mqtt.port');
        $this->timeout      = config('mqtt.timeout');
        $this->debug        = config('mqtt.debug');
        $this->qos          = config('mqtt.qos');
        $this->retain       = config('mqtt.retain');
        $this->clientId     = config('mqtt.clientId');
        $this->clean        = config('mqtt.clean');
        $this->PublishID    = config('mqtt.PublishID');
        $this->SubscribeID  = config('mqtt.SubscribeID');
    }


    /**
     * 发布
     */
    public function ConnectAndPublish($topic, $msg, $client_id=null, $retain=null)
    {
        $client     = new Client($this->host, $this->port, $this->getPublishConfig(), Client::COROUTINE_CLIENT_TYPE);
        $client->connect();
        while (true) {
            $response = $client->publish($topic, $msg, $this->qos);
            if ($response)
            {
                $client->close();
                return true;
            }
            $client->sleep(1);
        }
        return false;
    }

    function getPublishConfig()
    {
        $clientId   = $this->PublishID ? $this->PublishID : $this->clientId;
        $config     = new ClientConfig();

        return $config->setUserName($this->username) // 用户名
        ->setPassword($this->password) // 密码
        ->setClientId($clientId) // 客户端id
        ->setKeepAlive(10) // 默认0秒，设置成0代表禁用
        ->setDelay(3000) // 重连时的延迟时间 (毫秒)
        ->setMaxAttempts(5) // 最大重连次数。默认-1，表示不限制
        ->setProtocolName('MQTT') // 协议名，默认为MQTT(3.1.1版本)，也可为MQIsdp(3.1版本)
        ->setProtocolLevel(4); // 协议等级，MQTT3.1.1版本为4，5.0版本为5，MQIsdp为3
    }

    /**
     * 订阅
     */
    public function ConnectAndSubscribe($topic, $proc, $client_id=null)
    {
        $topics[$topic] = 0;
        $client     = new Client($this->host, $this->port, $this->getSubscribeConfig(), Client::COROUTINE_CLIENT_TYPE);
        $client->connect($this->clean);
        $client->subscribe($topics);
        $timeSincePing = time();
        while (true) {
            try {
                $buffer = $client->recv();
                if ($buffer && $buffer !== true) {
                    var_dump($buffer);
                    if (array_key_exists('topic', $buffer) && array_key_exists('message', $buffer))
                    {
                        $proc($buffer['topic'], $buffer['message']);
                    }
                    // QoS1 PUBACK
                    if ($buffer['type'] === Types::PUBLISH && $buffer['qos'] === 1) {
                        $client->send(
                            [
                                'type' => Types::PUBACK,
                                'message_id' => $buffer['message_id'],
                            ],
                            false
                        );
                    }
                    if ($buffer['type'] === Types::DISCONNECT) {
                        echo "Broker已断开连接\n";
                        $client->close();
                        break;
                    }
                }
                if ($timeSincePing <= (time() - $client->getConfig()->getKeepAlive())) {
                    $buffer = $client->ping();
                    if ($buffer) {
//                        echo 'send ping success' . PHP_EOL;
                        $timeSincePing = time();
                    }
                }
            } catch (\Throwable $e) {
                throw $e;
            }
        }
    }


    function getSubscribeConfig()
    {
        $clientId   = $this->SubscribeID ? $this->SubscribeID : $this->clientId;
        $config     = new ClientConfig();

        return $config->setUserName($this->username) // 用户名
        ->setPassword($this->password) // 密码
        ->setClientId($clientId) // 客户端id
        ->setKeepAlive(10) // 默认0秒，设置成0代表禁用
        ->setDelay(3000) // 重连时的延迟时间 (毫秒)
        ->setMaxAttempts(5) // 最大重连次数。默认-1，表示不限制
        ->setProtocolName('MQTT') // 协议名，默认为MQTT(3.1.1版本)，也可为MQIsdp(3.1版本)
        ->setProtocolLevel(4); // 协议等级，MQTT3.1.1版本为4，5.0版本为5，MQIsdp为3
    }
}
